package com.toonyoo.eurekaproducer.config;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.sun.xml.internal.messaging.saaj.util.ByteOutputStream;
import com.toonyoo.eurekaproducer.model.Student;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.SerializationException;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.net.UnknownHostException;

@Configuration
public class KryoSerializerConfiguration extends RedisAutoConfiguration {

    private ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(Kryo::new);

    @Override
    @Bean
    public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException {
        RedisTemplate myRedisTemplate = super.redisTemplate(redisConnectionFactory);
        // 自定义序列化策略
        myRedisTemplate.setKeySerializer(new StringRedisSerializer());
        myRedisTemplate.setValueSerializer(new RedisSerializer<Object>() {
            @Override
            public byte[] serialize(Object o) throws SerializationException {
                if(o == null){
                    return new byte[0];
                }
                Kryo kryo = kryoThreadLocal.get();
                ByteOutputStream bos = new ByteOutputStream();
                Output output = new Output(bos);
                kryo.writeObject(output, o);
                output.close();
                return bos.getBytes();
            }

            @Override
            public Object deserialize(byte[] bytes) throws SerializationException {
                if(bytes == null || bytes.length == 0){
                    return null;
                }
                Kryo kryo = kryoThreadLocal.get();
                Input input = new Input(bytes);
                return kryo.readObject(input, Object.class);
            }
        });
        return myRedisTemplate;
    }

    @Override
    @Bean
    public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException {
        StringRedisTemplate stringRedisTemplate = super.stringRedisTemplate(redisConnectionFactory);
        // TODO 自定义序列化策略
        return stringRedisTemplate;
    }

    @Bean
    public RedisTemplate<String, Student> myRedisTemplate(RedisConnectionFactory redisConnectionFactory){
        RedisTemplate myRedisTemplate = new RedisTemplate();
        myRedisTemplate.setConnectionFactory(redisConnectionFactory);
        myRedisTemplate.setKeySerializer(new StringRedisSerializer());
        myRedisTemplate.setValueSerializer(new RedisSerializer<Student>() {
            @Override
            public byte[] serialize(Student o) throws SerializationException {
                if(o == null){
                    return new byte[0];
                }
                Kryo kryo = kryoThreadLocal.get();
                ByteOutputStream bos = new ByteOutputStream();
                Output output = new Output(bos);
                kryo.writeObject(output, o);
                output.close();
                return bos.getBytes();
            }

            @Override
            public Student deserialize(byte[] bytes) throws SerializationException {
                if(bytes == null || bytes.length == 0){
                    return null;
                }
                Kryo kryo = kryoThreadLocal.get();
                Input input = new Input(bytes);
                return kryo.readObject(input, Student.class);
            }
        });

        return myRedisTemplate;
    }
}
